/**
 * @file   MzSequenceFile.java
 * @author Wu Jian <wujian@miaozhen.com>
 * @date   Wed Mar  6 15:49:04 2013
 *
 * @brief  Sequence File specific for miaozhen
 *
 *
 */

package com.miaozhen.utils;

import java.io.*;
import java.util.*;
import java.nio.charset.Charset;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.VersionMismatchException;
import org.apache.hadoop.io.WritableName;
import org.apache.hadoop.io.compress.CodecPool;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionInputStream;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.io.compress.Compressor;
import org.apache.hadoop.io.compress.Decompressor;
import org.apache.hadoop.io.serializer.Deserializer;
import org.apache.hadoop.io.serializer.SerializationFactory;
import org.apache.hadoop.io.serializer.Serializer;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.util.ReflectionUtils;


public class MzSequenceFile {
    private MzSequenceFile() {}/// no public constructors

    private static final byte LATEST_VERSON = (byte)1;
    private static byte[] VERSION = new byte[] {
            (byte)'M', (byte)'Z', (byte)'S', (byte)'E', (byte)'Q', LATEST_VERSON
    };
    private static final Charset CHARSET = Charset.forName("UTF-8");
    private static final byte LF = '\n';

    /**
     * @brief This class encapsulating with the metadata of a part.
     *        The metadata of a part is a pair of offset and length.
     *
     */
    public static class SMeta {
        public long offset;
        public long length;
        public long number;

        public SMeta() {
            this.offset = this.length = this.number = 0;
        }

        public SMeta(long offset, long length, long number) {
            this.offset = offset;
            this.length = length;
            this.number = number;
        }

        public void write(DataOutput out) throws IOException{
            out.writeLong(this.offset);
            out.writeLong(this.length);
            out.writeLong(this.number);
        }

        public void readFields(DataInput in) throws IOException{
            this.offset = in.readLong();
            this.length = in.readLong();
            this.number = in.readLong();
        }

        public boolean equals(SMeta other) {
            if (this.offset == other.offset && this.length == other.length
                    && this.number == other.number) {
                return true;
            }
            return false;
        }

        public String toString() {
            StringBuffer sb = new StringBuffer();
            return sb.append(Long.toString(this.offset)).append(',').append(Long.toString(this.length)).append(':').append(Long.toString(this.number)).toString();
        }
    }

    /**
     * @brief This class encapsulating with the metadata of a file.
     *        The metadata of a file is list of pairs of offset and length.
     *
     */
    public static class MetaData implements Writable {
        private TreeMap<String, SMeta> theMetadata;

        public MetaData() {
            this.theMetadata = new TreeMap<String, SMeta>();
        }

        public MetaData(TreeMap<String, SMeta> arg) {
            if (null == arg) {
                this.theMetadata = new TreeMap<String, SMeta>();
            } else {
                this.theMetadata = arg;
            }
        }

        public long getTotalNumber() {
            Collection<SMeta> c = theMetadata.values();
            Iterator<SMeta> itr = c.iterator();
            long totalNumber = 0;
            while (itr.hasNext()) {
                totalNumber += itr.next().number;
            }
            return totalNumber;
        }

        public SMeta get(String name) {
            return this.theMetadata.get(name);
        }

        public void set(String name, long offset, long length, long number) {
            this.theMetadata.put(name, new SMeta(offset, length, number));
        }

        public void set(String name, SMeta smeta) {
            this.theMetadata.put(name, smeta);
        }

        public TreeMap<String, SMeta> getMetadata() {
            return new TreeMap<String, SMeta>(this.theMetadata);
        }

        /**
         * @brief Write a String object to DataOutput,
         *        composed of length and bytes
         */
        private int writeString(DataOutput out, String s) throws IOException {
            byte[] buffer = s.getBytes(CHARSET);
            int length = buffer.length;
            out.writeInt(length);
            out.write(buffer, 0, length);
            return length+4;
        }

        private String readString(DataInput in) throws IOException {
            int length = in.readInt();
            byte[] buffer = new byte[length];
            in.readFully(buffer, 0, length);
            return new String(buffer, CHARSET);
        }

        public void write(DataOutput out) throws IOException {
            out.writeInt(this.theMetadata.size());
            Iterator<Map.Entry<String, SMeta>> iter =
                    this.theMetadata.entrySet().iterator();
            while (iter.hasNext()) {
                Map.Entry<String, SMeta> en = iter.next();
                writeString(out, en.getKey());
                en.getValue().write(out);
            }
        }

        public void readFields(DataInput in) throws IOException {
            int sz = in.readInt();
            if (sz < 0) throw new IOException("Invalid size: " + sz + " for file metadata object");
            this.theMetadata = new TreeMap<String, SMeta>();
            for (int i = 0; i < sz; i++) {
                String key = readString(in);
                SMeta val = new SMeta();
                val.readFields(in);
                this.theMetadata.put(key, val);
            }
        }

        public boolean equals(MetaData other) {
            if (other == null) return false;
            if (this.theMetadata.size() != other.theMetadata.size()) {
                return false;
            }
            Iterator<Map.Entry<String, SMeta>> iter1 =
                    this.theMetadata.entrySet().iterator();
            Iterator<Map.Entry<String, SMeta>> iter2 =
                    other.theMetadata.entrySet().iterator();
            while (iter1.hasNext() && iter2.hasNext()) {
                Map.Entry<String, SMeta> en1 = iter1.next();
                Map.Entry<String, SMeta> en2 = iter2.next();
                if (!en1.getKey().equals(en2.getKey())) {
                    return false;
                }
                if (!en1.getValue().equals(en2.getValue())) {
                    return false;
                }
            }
            if (iter1.hasNext() || iter2.hasNext()) {
                return false;
            }
            return true;
        }

        public int hashCode() {
            assert false : "hashCode not designed";
            return 42; // any arbitrary constant will do
        }

        public String toString() {
            StringBuffer sb = new StringBuffer();
            sb.append("size: ").append(this.theMetadata.size()).append("\n");
            Iterator<Map.Entry<String, SMeta>> iter =
                    this.theMetadata.entrySet().iterator();
            while (iter.hasNext()) {
                Map.Entry<String, SMeta> en = iter.next();
                sb.append("\t").append(en.getKey().toString()).append("\t").append(en.getValue().toString());
                sb.append("\n");
            }
            return sb.toString();
        }
    }

    /** Write logs to a sequence-format file. */
    public static class Writer implements java.io.Closeable {
        Configuration conf;
        FSDataOutputStream out;
        boolean ownOutputStream = true;
        private String currentMeta = null;
        private long lastPos = 0;

        Class valClass;

        private boolean compress;
        MetaData metadata = new MetaData();
        CompressionCodec codec = null;
        CompressionOutputStream deflateFilter = null;
        DataOutputStream deflateOut = null;
        Compressor compressor = null;

        protected Serializer valSerializer;
        public long number = 0;

        public Writer(FileSystem fs, Configuration conf, Path path,
                      Class valClass) throws IOException {
            this.init(conf, fs.create(path), valClass, true, null);
        }

        public Writer(FileSystem fs, Configuration conf, Path path,
                      Class valClass, CompressionCodec codec)
                throws IOException {
            this.init(conf, fs.create(path), valClass, true, codec);
        }

        public Writer(Configuration conf, FSDataOutputStream out,
                      Class valClass) throws IOException {
            this.init(conf, out, valClass, false, null);
        }

        public void init(Configuration conf, FSDataOutputStream out,
                         Class valClass, boolean ownOutputStream,
                         CompressionCodec codec)
                throws IOException {
            this.conf = conf;
            this.compress = (codec == null) ? false:true;
            this.out = out;
            this.ownOutputStream = ownOutputStream;
            this.valClass = valClass;
            SerializationFactory serializationFactory = new SerializationFactory(conf);
            this.valSerializer = serializationFactory.getSerializer(valClass);
            this.codec = codec;
            if (codec == null) {
                this.valSerializer.open(this.out);
            } else {
                ReflectionUtils.setConf(this.codec, this.conf);
                this.compressor = CodecPool.getCompressor(this.codec);
                this.deflateFilter = this.codec.createOutputStream(this.out, compressor);
                this.deflateOut =
                        new DataOutputStream(new BufferedOutputStream(deflateFilter));
                this.valSerializer.open(this.deflateOut);
            }
        }

        /** Append a value with meta. */
        public synchronized void append(String meta, Writable val)
                throws IOException {
            this.append(meta, (Object) val);
        }

        /** Append a value with meta. */
        public synchronized void append(String meta, Object val)
                throws IOException {
            if (this.currentMeta == null)
                this.setMeta(meta);
            if (!meta.equals(this.currentMeta)) {
                this.sync();
                this.setMeta(meta);
            }
            this.append(val);
        }

        /** Append a value. */
        public synchronized void append(Writable val)
                throws IOException {
            append((Object) val);
        }

        /** Append a value. */
        public synchronized void append(Object val)
                throws IOException {
            if (val.getClass() != valClass)
                throw new IOException("wrong value class: "+val.getClass().getName()
                        +" is not "+valClass);

            valSerializer.serialize(val);
            ++this.number;
        }

        public synchronized void appendRaw(byte[] val, int offset, int length)
                throws IOException {
            if (length < 0)
                throw new IOException("negative length values not allowed: " + length);
            this.out.write(val, offset, length);
            ++this.number;
        }

        public void setMeta(String meta) {
            this.currentMeta = meta;
        }

        public synchronized void sync() throws IOException {
            if (this.currentMeta == null) {
                throw new IOException("Not set meta yet!");
            }
            if (this.metadata.get(this.currentMeta) != null) {
                throw new IOException("The same meta added!");
            }
            if (this.isCompressed()) {
                this.deflateOut.flush();
                this.deflateFilter.finish();
                CodecPool.returnCompressor(compressor);
                compressor = null;
            }
            long currentPos = this.out.getPos();
            this.metadata.set(this.currentMeta, this.lastPos,
                    currentPos-this.lastPos, this.number);
            this.lastPos = currentPos;
            this.currentMeta = null;
            this.number = 0;
            if (this.isCompressed()){
                this.compressor = CodecPool.getCompressor(this.codec);
                this.deflateFilter = this.codec.createOutputStream(this.out, compressor);
                this.deflateOut =
                        new DataOutputStream(new BufferedOutputStream(deflateFilter));
                this.valSerializer.open(this.deflateOut);
            }
        }

        private void writeTailer() throws IOException {
            this.lastPos = this.out.getPos();
            this.out.write(VERSION, 0, VERSION.length);
            Text.writeString(this.out, valClass.getName());
            this.out.writeBoolean(this.compress);
            if (this.isCompressed())
                Text.writeString(this.out, (codec.getClass()).getName());
            this.metadata.write(this.out);
            long currentPos = this.out.getPos();
            this.out.writeInt((int)(currentPos-this.lastPos));
        }

        boolean isCompressed() { return this.compress; }

        public synchronized void close() throws IOException {
            if (this.currentMeta != null) {
                this.sync();
            }

            CodecPool.returnCompressor(compressor);
            compressor = null;

            this.writeTailer();

            if (this.ownOutputStream) {
                this.out.close();
            } else {
                this.out.flush();
            }
            this.out = null;
        }
    }

    /** Read a part from a FSDataInputStream */
    public static class PartInputStream extends java.io.InputStream {
        private FSDataInputStream in;
        private long start;
        private long end;

        public PartInputStream(FSDataInputStream in) {
            this.in = in;
            this.start = this.end = 0;
        }

        public PartInputStream(FSDataInputStream in,
                               long start, long end)
                throws IOException {
            this.in = in;
            this.start = start;
            this.end = end;
            this.in.seek(start);
        }

        public int read() throws IOException {
            if (in.getPos() < end) {
                return in.read();
            }
            return -1;
        }

        private int min(int a, int b) {
            return (a < b) ? a : b;
        }

        public int read(byte[] b, int off, int len) throws IOException {
            if (in.getPos() < end) {
                long clen = Math.min((long)len, end - in.getPos());
                if (clen > ((long) Integer.MAX_VALUE)) {
                    throw new IOException(String.format("clen :%s is longer than :%s", clen, Integer.MAX_VALUE));
                }
                return in.read(b, off, (int)clen);
            }
            return -1;
        }


        public int read(byte[] b) throws IOException {
            return read(b, 0, b.length);
        }

        public int available() throws IOException {
            return in.available();
        }

        public void close() throws IOException {
            in.close();
        }

        public void mark(int readlimit) {
            in.mark(readlimit);
        }

        public void reset() throws IOException {
            in.reset();
        }

        public boolean markSupported() {
            return in.markSupported();
        }

        public long getStart() {
            return this.start;
        }

        public long getEnd() {
            return this.end;
        }

        public long getPos() throws IOException {
            return this.in.getPos();
        }
    }

    /** Read logs to a sequence-format file. */
    public static class Reader implements java.io.Closeable {
        private Configuration conf;
        private FSDataInputStream in;
        private PartInputStream partIn;

        private byte version;
        private MetaData metadata = null;
        private String meta;

        private long end;
        private long length;
        private long contentEnd;
        private long number = 0;
        private long readNumber = 0;

        private String valClassName;
        private Class valClass;

        private boolean decompress;
        private CompressionCodec codec = null;
        private CompressionInputStream valInFilter = null;
        private DataInputStream valIn = null;
        private Decompressor valDecompressor = null;

        private Deserializer valDeserializer;

        public Reader(FileSystem fs, Path path, long length,
                      Configuration conf, String meta)
                throws IOException {
            this(fs.open(path), length, conf, meta);
        }

        public Reader(FSDataInputStream in, long length,
                      Configuration conf, String meta)
                throws IOException {
            this.in = in;
            this.partIn = new PartInputStream(in);
            this.conf = conf;
            this.length = length;
            init();
            setMeta(meta);
        }

        private void init() throws IOException {
            this.in.seek(this.length-4);
            int tailLength = this.in.readInt();
            this.contentEnd = this.length-4-tailLength;
            this.in.seek(this.contentEnd);

            byte[] versionBlock = new byte[VERSION.length];
            this.in.readFully(versionBlock);

            if ((versionBlock[0] != VERSION[0]) ||
                    (versionBlock[1] != VERSION[1]) ||
                    (versionBlock[2] != VERSION[2]) ||
                    (versionBlock[3] != VERSION[3]) ||
                    (versionBlock[4] != VERSION[4]))
                throw new IOException("Input is not a MzSequenceFile");

            // Set 'version'
            this.version = versionBlock[5];
            if (version > VERSION[5])
                throw new VersionMismatchException(VERSION[3], version);

            this.valClassName = Text.readString(this.in); // val class name

            this.decompress = this.in.readBoolean();
            if (this.decompress){
                String codecClassname = Text.readString(this.in);
                try {
                    Class<? extends CompressionCodec> codecClass
                            = conf.getClassByName(codecClassname).asSubclass(CompressionCodec.class);
                    this.codec = ReflectionUtils.newInstance(codecClass, conf);
                } catch (ClassNotFoundException cnfe) {
                    throw new IllegalArgumentException("Unknown codec: " +
                            codecClassname, cnfe);
                }
            }

            this.metadata = new MetaData();
            this.metadata.readFields(this.in);

            SerializationFactory serializationFactory =
                    new SerializationFactory(conf);
            this.valDeserializer =
                    serializationFactory.getDeserializer(getValueClass());
        }

        public long getStart() {
            return this.partIn.getStart();
        }

        public long getEnd() {
            return this.partIn.getEnd();
        }

        public long getPos() throws IOException {
            return this.partIn.getPos();
        }

        /** Returns the name of the value class. */
        public String getValueClassName() {
            return valClassName;
        }

        /** Returns the class of values in this file. */
        public synchronized Class<?> getValueClass() {
            if (null == valClass) {
                try {
                    valClass = WritableName.getClass(getValueClassName(), conf);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }
            return valClass;
        }

        public void setMeta(String meta) throws IOException {
            this.meta = meta;
            if (meta == null) {
                this.partIn = new PartInputStream(this.in, 0,
                        this.contentEnd);
                this.number = this.metadata.getTotalNumber();
            } else {
                SMeta smeta = this.metadata.get(meta);
                if (smeta != null) {
                    this.partIn = new PartInputStream(this.in, smeta.offset,
                            smeta.offset+smeta.length);
                    this.number = smeta.number;
                } else {
                    this.partIn = new PartInputStream(this.in);
                    this.number = 0;
                }
            }
            this.readNumber = 0;
            if (this.isCompressed()) {
                CodecPool.returnDecompressor(valDecompressor);
                valDecompressor = null;
                valDecompressor = CodecPool.getDecompressor(codec);
                valInFilter = codec.createInputStream(this.partIn,
                        valDecompressor);
                valIn = new DataInputStream(valInFilter);
                this.valDeserializer.open(this.valIn);
            } else {
                this.valDeserializer.open(this.partIn);
            }
        }

        /** Returns true if values are compressed. */
        public boolean isCompressed() { return decompress; }

        /** Returns the configuration used for this file. */
        Configuration getConf() { return conf; }

        /** Returns the metadata object of the file */
        public MetaData getMetadata() {
            return this.metadata;
        }

        private Object deserializeValue(Object val) throws IOException {
            return valDeserializer.deserialize(val);
        }

        public synchronized boolean next(Object val)
                throws IOException {
            if (val.getClass() != getValueClass())
                throw new IOException("wrong value class: "+val+" is not "+valClass);
            if (this.readNumber >= this.number)
                return false;
            val = deserializeValue(val);
            ++this.readNumber;
            return true;
        }

        public synchronized boolean next(Writable val)
                throws IOException {
            return next((Object) val);
        }

        public void close() throws IOException {
            CodecPool.returnDecompressor(valDecompressor);
            valDecompressor = null;

            if (this.valDeserializer != null) {
                this.valDeserializer.close();
            }
            this.in.close();
        }
    }
}